package com.atguigu.spark.core.wc

import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.rdd.RDD
import org.apache.spark.util.AccumulatorV2

import scala.Console.println
import scala.collection.mutable

object WordCount {
  def main(args: Array[String]): Unit = {
    val conf: SparkConf = new SparkConf().setMaster("local").setAppName("WordCount")
    conf .set("spark.testing.memory","2147480000")
    val sc: SparkContext = new SparkContext(conf)
    val list: RDD[String] = sc.textFile("spark-core/data/word.txt")
    val words: RDD[String] = list.flatMap(_.split(" "))
    val wordToOne: RDD[(String, Int)] = words.map((_, 1))
    //一
    println(wordToOne.groupBy(_._1).map(x => (x._1, x._2.size)).collect().mkString(" , "))
    //二
    println(wordToOne.groupByKey().map(x => (x._1, x._2.size)).collect().mkString(" , "))
    //三
    println(wordToOne.reduceByKey(_ + _).collect().mkString(" , "))
    //四
    println(wordToOne.aggregateByKey(0)(_ + _, _ + _).collect().mkString(" , "))
    //五
    println(wordToOne.foldByKey(0)(_ + _).collect().mkString(" , "))
    //六
    println(wordToOne.countByKey().mkString(" , "))
    //七
    println(words.countByValue().mkString(" , "))
    //八
    println(wordToOne.combineByKey(x => x, (left: Int, right: Int) => {left + right}, (left: Int, right: Int) => {left + right}).collect().mkString(" , "))
    //九
    println(wordToOne.aggregate(Map[String, Int]())(
      (map1, kv) => {
        val k = kv._1
        val v = kv._2
        val newvalue = map1.getOrElse(k, 0) + v
        map1.updated(k, newvalue)
      },
      (map1, map2) => {
        map1.foldLeft(map2)((map, kv) => {
          val k: String = kv._1
          val v: Int = kv._2
          val newValue = map.getOrElse(k, 0) + v
          map.updated(k, newValue)
        })
      }
    ).mkString(" , "))
    //十
    val acc: MyAcc = new MyAcc
    sc.register(acc,"wordCount")
    words.foreach(x=>acc.add(x))
    println(acc.value)


  }

}
class MyAcc extends AccumulatorV2[String,mutable.Map[String,Long]]{
  var map:mutable.Map[String,Long] = mutable.Map()
  override def isZero: Boolean = {
    map.isEmpty
  }

  override def copy(): AccumulatorV2[String, mutable.Map[String, Long]] = {
    new MyAcc
  }

  override def reset(): Unit = {
    map.clear()
  }

  override def add(word: String): Unit = {
    val old: Long = map.getOrElse(word, 0)
    map.update(word,old+1)
  }

  override def merge(other: AccumulatorV2[String, mutable.Map[String, Long]]): Unit = {
    val map2: mutable.Map[String, Long] = other.value
    map2.foreach {
      case (word, count) => {
        val oldCnt: Long = map.getOrElse(word, 0)
        map.update(word, oldCnt + count)
      }
    }
  }

  override def value: mutable.Map[String, Long] = {
    map
  }
}